﻿using System;
using Confluent.Kafka;
using Microsoft.Extensions.Options;

namespace Micro.Extensions.Log.Kafka
{
    public class KafkaClient : IKafkaClient
    {
        private  IProducer<Null, string> _producer;
        private readonly IOptionsMonitor<KafkaClientConfig> _options;
        private readonly object _lockObj = new object();
        private readonly IDisposable _cancelDisposable;
        public KafkaClient(IOptionsMonitor<KafkaClientConfig> options)
        {
            _options = options;
            _cancelDisposable = _options.OnChange(ReBuildProducer);
        }

        public IProducer<Null, string> Producer()
        {
            if (_producer!=null)
            {
                return _producer;

            }
            lock (_lockObj)
            {
                if (_producer != null)
                {
                    return _producer;

                }
                _producer = BuildProducer();
            }
            return _producer;
        }
        /// <summary>
        /// 构建Producer
        /// </summary>
        private IProducer<Null, string> BuildProducer()
        {
            return new ProducerBuilder<Null, string>(new ProducerConfig
            {
                BootstrapServers = _options.CurrentValue.BootstrapServers,
                SaslMechanism = (SaslMechanism)_options.CurrentValue.SaslMechanism,
                SecurityProtocol = (SecurityProtocol)_options.CurrentValue.SecurityProtocol,                
                SaslUsername= _options.CurrentValue.SaslUsername,
                SaslPassword = _options.CurrentValue.SaslPassword
                
            }).Build();
        }

        //根据配置构建新的Producer
        private void ReBuildProducer(KafkaClientConfig option)
        {
            var oldProducer = _producer;
            _producer = BuildProducer();

            //强制缓存数据推送到Broker
            oldProducer.Flush();
            //释放资源
            oldProducer.Dispose();
        }

        public void Dispose()
        {
            //释放kafka资源
            _producer.Flush();
            _producer.Dispose();
            _cancelDisposable.Dispose();
        }
    }
}
